package com.jial08.rabbitmqretry.component;

import com.jial08.rabbitmqretry.domain.QueueConstant;
import com.jial08.rabbitmqretry.domain.QueueEnum;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * <p>类描述：消费消息。</p>
 *
 * @author jl
 * @version v1.0.0.1。
 * @since JDK1.8。
 * <p>创建日期：2020/12/8 下午10:32。</p>
 */
@Component
@Slf4j
public class Consumer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = "mall.pay.topic")
    public void sendMail(Message message, Map<String, Object> map, Channel channel) throws IOException {
        String correlationId = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        try {
            log.warn("message：{}", map);
            // 手动抛出异常,测试消息重试
            int i = 1 / 0;
        } catch (Exception e) {
            CorrelationData correlationData = new CorrelationData(correlationId);
            long retryCount = getRetryCount(message.getMessageProperties());
            if (retryCount >= QueueConstant.RETRY_TIMES) {
                // 重试次数超过3次,则将消息发送到失败队列等待特定消费者处理或者人工处理
                try {
                    rabbitTemplate.convertAndSend(QueueEnum.QUEUE_ORDER_PAY_FAILD.getExchange(),
                            QueueEnum.QUEUE_ORDER_PAY_FAILD.getRouteKey(), message, correlationData);
                } catch (Exception e1) {
                    log.warn("消息在发送到 failed 队列的时候报错：{}，原始消息：{}", e1.getMessage(), message);
                }
            } else {
                try {
                    // 重试次数不超过3次,则将消息发送到重试队列等待重新被消费（重试队列延迟超时后信息被发送到相应死信队列重新消费，即延迟消费）
                    rabbitTemplate.convertAndSend(QueueEnum.QUEUE_ORDER_PAY_RETRY.getExchange(),
                            QueueEnum.QUEUE_ORDER_PAY_RETRY.getRouteKey(), message, correlationData);
                    log.warn("消费者消费失败，消息发送到重试队列；" + "原始消息：{}；第 {} 次重试。", message, retryCount + 1);
                } catch (Exception e1) {
                    log.warn("消息发送到重试队列时异常：{}", e1.getMessage());
                }
            }
        } finally {
            /**
             * 无论消费成功还是消费失败，都要手动进行 ack，因为即使消费失败了，也已经将消息重新投递到重试队列或者失败队列
             * 如果不进行 ack，生产者在超时后会进行消息重发，如果消费者依然不能处理，则会存在死循环
             */
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    /**
     * 获取消息被重试的次数
     */
    public long getRetryCount(MessageProperties messageProperties) {
        Long retryCount = 0L;
        if (null != messageProperties) {
            List<Map<String, ?>> deaths = messageProperties.getXDeathHeader();
            if (deaths != null && deaths.size() > 0) {
                Map death = deaths.get(0);
                retryCount = (Long) death.get("count");
            }
        }
        return retryCount;
    }
}
